package com.atbeijing.D03;

import com.atbeijing.D02.SensorReading;
import com.atbeijing.D02.SensorSource;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;


/**
 * 自定义窗口聚合函数,处理滚动时间窗口数据
 * 窗口数据增量聚合:当一个事件进入窗口时触发聚合计算
 *
 * 求5s内每个温感器的平均温度
 */
public class Example7 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        env
                .addSource(new SensorSource())
                .keyBy(r -> r.id)
                .window(TumblingProcessingTimeWindows.of(Time.seconds(5)))//每个温感器开一个5s的滚动处理时间窗口
                .aggregate(new Agg())//自定义增量聚合函数,适用于每条流的窗口
                .print();
//         统计结果:每个温控器,每隔5秒计算一次平均温度
//        (sensor_0,70.46232063971468)
//        (sensor_9,85.16656220280241)
//        (sensor_8,37.18358332687101)
//        (sensor_7,66.56046958018447)
//        (sensor_6,48.95898357836333)
//        (sensor_5,85.6349400832225)
//        (sensor_4,55.48150087743579)
//        (sensor_3,42.872831980804484)
//        (sensor_2,28.669196920528165)
//        (sensor_1,75.82684022023655)
        env.execute();
    }

    /**
     * 自定义增量聚合
     * AggregateFunction的泛型
     *  <IN>  输入类型 SensorReading
     *  <ACC> 累加器类型 Tuple3
     *  <OUT> 窗口结束后下发的数据类型 Tuple2
     */
    public static class Agg implements AggregateFunction<SensorReading, Tuple3<String, Double, Long>, Tuple2<String, Double>>{

        //初始化累加器
        @Override
        public Tuple3<String, Double, Long> createAccumulator() {
            return Tuple3.of("",0.0,0L);
        }

        //窗口中每来一个事件触发一次累加
        //value: 窗口的一个事件
        //accumulator: 累加器
        @Override
        public Tuple3<String, Double, Long> add(SensorReading value, Tuple3<String, Double, Long> accumulator) {
            //温感器id
            accumulator.f0=value.id;
            //温度累加
            accumulator.f1=accumulator.f1+value.temperature;
            //每来一条数据+1
            accumulator.f2=accumulator.f2+1L;
            return accumulator;
        }

        //窗口闭合后向下游发送的数据,在这一步处理累加器的值
        @Override
        public Tuple2<String, Double> getResult(Tuple3<String, Double, Long> accumulator) {
            //(温感器id,温度总和/数据条数)
            return Tuple2.of(accumulator.f0,accumulator.f1/accumulator.f2);
        }

        @Override
        public Tuple3<String, Double, Long> merge(Tuple3<String, Double, Long> a, Tuple3<String, Double, Long> b) {
            return null;
        }
    }

}
